package com.nucarf.bi.digtalize.function;

import com.nucarf.bi.digtalize.bean.PushConsume;
import com.nucarf.bi.digtalize.bean.SinkKafkaBean;
import org.apache.commons.collections.CollectionUtils;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

public class ProcessOrderFunction extends KeyedProcessFunction<String, SinkKafkaBean, SinkKafkaBean> {

    private MapState<Long, List> mapState;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        mapState = getRuntimeContext().getMapState(
                new MapStateDescriptor<>("updateOrderState", TypeInformation.of(Long.class), TypeInformation.of(List.class))
        );
    }

    @Override
    public void processElement(SinkKafkaBean sinkKafkaBean, Context context, Collector<SinkKafkaBean> collector) throws Exception {
        if (sinkKafkaBean.getSendBean() instanceof PushConsume) {
            PushConsume pushConsume = (PushConsume) sinkKafkaBean.getSendBean();
            Long time = System.currentTimeMillis() ;
            if (pushConsume.getChange_time().longValue() != 0 || pushConsume.getUpdate_time().longValue() != 0) {
                context.timerService().registerProcessingTimeTimer(time + 5000L);
                List list = mapState.get(time);
                if (CollectionUtils.isEmpty(list)) {
                    list = new ArrayList<SinkKafkaBean>();
                }
                list.add(sinkKafkaBean);
                mapState.put(time, list);
            } else {
                collector.collect(sinkKafkaBean);
            }
        } else {
            collector.collect(sinkKafkaBean);
        }
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<SinkKafkaBean> out) throws Exception {
        Iterable<Map.Entry<Long, List>> entries = mapState.entries();
        Iterator<Map.Entry<Long, List>> iterator = entries.iterator();
        List<Long> removes = new ArrayList<>();
        while (iterator.hasNext()) {
            Map.Entry<Long, List> next = iterator.next();
            Long time = next.getKey();
            if (timestamp - time > 4000L) {
                List<SinkKafkaBean> value = next.getValue();
                for (SinkKafkaBean sinkBean : value) {
                    out.collect(sinkBean);
                }
                removes.add(time);
            }
        }
        if (!CollectionUtils.isEmpty(removes)) {
            for (Long key : removes) {
                mapState.remove(key);
            }
        }
    }
}
